package com.mingserve.raw.listener;

import com.alibaba.fastjson2.JSON;
import com.mingserve.common.core.constant.CacheConstants;
import com.mingserve.common.core.constant.RawConstants;
import com.mingserve.common.core.text.Convert;
import com.mingserve.common.core.utils.*;
import com.mingserve.common.mqtt.utils.MqttUtils;
import com.mingserve.common.redis.service.RedisService;
import com.mingserve.raw.domain.*;
import com.mingserve.raw.domain.dto.RawConnDto;
import com.mingserve.raw.domain.dto.RawSubDataDto;
import com.mingserve.raw.event.RawEvent;
import com.mingserve.raw.service.IRawService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 订阅PublicEvent中的传感器数据
 *
 * @author lvwshuai
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class RawListener {
    private final RedisService redisService;
    private final MqttUtils mqttUtils;
    private final IRawService rawService;
    private static final String UPLOAD_SPEED = "UploadSpeed";

    /**
     * 攒包等待超时时间，默认20分钟
     */
    @Value("${spring.rawJson.wait-timeout:20}")
    private long waitTimeout;
    @Value("${spring.rawJson.trans-mqtt:false}")
    private boolean transMqtt;
    @Value("${spring.rawJson.write:true}")
    private boolean write;
    @Value("${spring.rawJson.trans-api:false}")
    private boolean transApi;
    @Value("${spring.rawJson.trans-url}")
    private String transUrl;

    @EventListener(RawEvent.class)
    public void rawEventListener(RawEvent rawEvent) {
        log.info("Start unpacking {}", rawEvent.getSource());
        switch (rawEvent.getDataType()) {
            case RawConstants.KAFKA_RAW_DATA_TOPIC -> unpack(rawEvent.getMap());
            case RawConstants.KAFKA_RAW_ZIP_TOPIC -> {
                // ZLib解压缩
                String data = ZipUtils.unZlibToStr(rawEvent.getBytes());
                Map<String, Object> map = JSON.parseObject(data);
                unpackBase64(map);
            }
            case RawConstants.KAFKA_RAW_SUB_ZIP_TOPIC -> {
                // ZLib解压缩
                String data = ZipUtils.unZlibToStr(rawEvent.getBytes());
                Map<String, Object> map = JSON.parseObject(data);
                unpackSubBase64(map);
            }
            case RawConstants.KAFKA_RAW_UNZIP_TOPIC -> unpackBase64(rawEvent.getMap());
            case RawConstants.KAFKA_RAW_SUB_UNZIP_TOPIC -> unpackSubBase64(rawEvent.getMap());
            case RawConstants.KAFKA_RAW_PRM_TOPIC -> unpackRPM(rawEvent.getMap());
        }
    }

    /**
     * 校验数据包ChipID+Timestamp
     *
     * @param resultMap
     * @return
     */
    private String checkPackage(Map<String, Object> resultMap) {
        if (resultMap == null || resultMap.isEmpty()) {
            return "Package is empty";
        }
        if (!resultMap.containsKey(RawConstants.CHIP_ID) || !resultMap.containsKey(RawConstants.TIMESTAMP)) {
            return "Invalid package, not found CHipID or Timestamp";
        }
        return null;
    }

    /**
     * 校验数据包
     * 1、ChipID+Timestamp不可为空
     * 2、判断Timestamp格式
     * 3、判断ChipID是否录入
     *
     * @param resultMap 成功：返回时间戳；失败：返回null
     */
    Long validPackage(Map<String, Object> resultMap) {
        String msg = checkPackage(resultMap);
        if (msg != null) {
            log.error("Package valid failed：{}", msg);
            return null;
        }
        String chipId = resultMap.get(RawConstants.CHIP_ID).toString();
        Long timestamp = DateUtils.formatTimestamp(resultMap.get(RawConstants.TIMESTAMP));
        if (timestamp == null) {
            log.error("Timestamp conversion failed：{}", chipId + "-" + resultMap.get(RawConstants.TIMESTAMP));
            if (resultMap.containsKey(UPLOAD_SPEED)) {
                refreshConfig(resultMap, false);
            }
            return null;
        }
        String title = chipId + "-" + timestamp;
        // 判断IoT设备是否已录入
        List<String> findPoint = queryPointListByDevice(chipId);
        if (findPoint == null || findPoint.isEmpty()) {
            log.info("{} not entered，not stored", title);
//            writeJson(resultMap);
            return null;
        }
        log.info("{} unpack successful, start storage", title);
        return timestamp;
    }

    /**
     * 标准解析 未压缩数据包、统计包
     *
     * @param resultMap
     */
    void unpack(Map<String, Object> resultMap) {
        Long timestamp = validPackage(resultMap);
        if (timestamp == null) return;
        resultMap.replace(RawConstants.TIMESTAMP, timestamp);
        insertRaw(resultMap);
    }

    /**
     * 未压缩的base64数据包-整包
     *
     * @param map
     */
    void unpackBase64(Map<String, Object> map) {
        boolean hasGain = map.containsKey(RawConstants.GAIN);
        // x y z 进行Base64解码
        for (String property : propertyArray) {
            processMapEntry(map, property, hasGain);
        }
        // 标准数据包解析
        unpack(map);
    }

    /**
     * base64解码出byte[] 后转 int[] 或 float[]
     *
     * @param map
     * @param key
     * @param hasGain
     */
    private void processMapEntry(Map<String, Object> map, String key, boolean hasGain) {
        if (map.containsKey(key)) {
            Object value = map.get(key);
            if (hasGain) {
                map.replace(key, ByteUtils.toIntArray(Base64Utils.decode(String.valueOf(value))));
            } else {
                map.replace(key, ByteUtils.toFloatArray(Base64Utils.decode(String.valueOf(value))));
            }
        }
    }

    /**
     * 未压缩的base64数据包-分包
     *
     * @param map
     */
    void unpackSubBase64(Map<String, Object> map) {
        String msg = checkPackage(map);
        if (msg != null) {
            log.error("Subpackage valid failed：{}", msg);
            return;
        }
        RawSubDataDto rawSubDataDto = JSON.parseObject(JSON.toJSONString(map), RawSubDataDto.class);
        String key = CacheConstants.RAW_SUBPACKAGE_CHIP + rawSubDataDto.getChipID() + "-" + rawSubDataDto.getTimestamp();
        List<RawSubDataDto> findList = redisService.getCacheList(key);
        boolean flag = false;
        if (findList == null) {
            findList = new ArrayList<>();
            findList.add(rawSubDataDto);
            flag = true;
        } else {
            // 判断是否已存在该包，不存在时攒包
            for (String property : propertyArray) {
                String rawData = getProperty(rawSubDataDto, property);
                if (rawData != null && findList.stream().noneMatch(l -> getProperty(l, property) != null && l.getSubPackageNum().equals(rawSubDataDto.getSubPackageNum()))) {
                    findList.add(rawSubDataDto);
                    flag = true;
                }
            }
        }
        // 分包收集中
        if (flag) {
            redisService.setCacheList(key, rawSubDataDto);
            redisService.expire(key, 6L, TimeUnit.MINUTES);
            log.info(key + " 攒包中，长度 " + findList.size());
        }
        RawSubDataDto find = findList.stream().filter(l -> l.getTotalSubPackage() != null && l.getGain() != null).findFirst().orElse(null);
        // 分包收集完毕
        // 2023年12月21日 带有FftAccX，送送速度数据，长度需*6，否则只有加速度数据，长度*3即可
        boolean isOK = find != null && ((find.getAccFeatureX() == null && findList.size() == find.getTotalSubPackage() * 3) || find.getAccFeatureX() != null && findList.size() == find.getTotalSubPackage() * 6);
        if (isOK) {
            log.info(key + " 攒包完毕，开始解析");
            Map<String, Object> resultMap = new HashMap<>();
            resultMap.put(RawConstants.CHIP_ID, find.getChipID());
            resultMap.put(RawConstants.TIMESTAMP, find.getTimestamp());
            resultMap.put(RawConstants.VIB_SAMPLE_FREQUENCY, find.getVibSampleFrequency());
            resultMap.put(RawConstants.OBJECT_TEMPERATURE, find.getObjectTemperature());
            boolean hasGain = !find.getGain().equals(0);
            if (hasGain) {
                resultMap.put(RawConstants.GAIN, find.getGain());
            }
            int size = 3;
            if (find.getAccFeatureX() != null) {
                size = 6;
                resultMap.put("FftAccX", find.getAccFeatureX());
                resultMap.put("FftAccY", find.getAccFeatureY());
                resultMap.put("FftAccZ", find.getAccFeatureZ());
                resultMap.put("FftVelX", find.getVelFeatureX());
                resultMap.put("FftVelY", find.getVelFeatureZ());
                resultMap.put("FftVelZ", find.getVelFeatureZ());
            }
            for (int i = 0; i < size; i++) {
                String property = propertyArray[i];
                processVibList(findList, property, resultMap, hasGain);
            }
            // 解标准包
            unpack(resultMap);
            redisService.deleteObject(key);
        }
    }

    private void processVibList(List<RawSubDataDto> dataList, String property, Map<String, Object> resultMap, boolean hasGain) {
        List<RawSubDataDto> filteredList = dataList.stream()
                .filter(item -> getProperty(item, property) != null)
                .sorted(Comparator.comparingInt(RawSubDataDto::getSubPackageNum))
                .toList();

        if (!filteredList.isEmpty()) {
            if (hasGain) {
                int[] rawArray = null;
                for (RawSubDataDto item : filteredList) {
                    String rawData = getProperty(item, property);
                    if (rawData != null) {
                        rawArray = ArrayUtils.addAll(rawArray, ByteUtils.toIntArray(Base64Utils.decode(rawData)));
                    }
                }
                resultMap.put(property, rawArray);
            } else {
                float[] rawArray = null;
                for (RawSubDataDto item : filteredList) {
                    String rawData = getProperty(item, property);
                    if (rawData != null) {
                        rawArray = ArrayUtils.addAll(rawArray, ByteUtils.toFloatArray(Base64Utils.decode(rawData)));
                    }
                }
                resultMap.put(property, rawArray);
            }
        }
    }

    /**
     * acc、vel 三向六个通道
     */
    private static final String[] propertyArray = {"accX", "accY", "accZ", "velX", "velY", "velZ"};

    private String getProperty(RawSubDataDto item, String property) {
        return switch (property) {
            case "accX" -> item.getAccX();
            case "accY" -> item.getAccY();
            case "accZ" -> item.getAccZ();
            case "velX" -> item.getVelX();
            case "velY" -> item.getVelY();
            case "velZ" -> item.getVelZ();
            default -> null;
        };
    }

    /**
     * 输出原始数据
     *
     * @param resultMap
     */
    private void insertRaw(Map<String, Object> resultMap) {
        // 2023-05-11 避免传感器无统计包
        String infoKey = CacheConstants.RAW_SENSOR_INFO_KEY + resultMap.get(RawConstants.CHIP_ID);
        Object old = redisService.getCacheMapValue(infoKey, RawConstants.EVENT_TIME);
        // 2023-08-24 最后上传时间取大
        if (old == null || Convert.toLong(old) < Convert.toLong(resultMap.get(RawConstants.TIMESTAMP))) {
            redisService.setCacheMapValue(infoKey, RawConstants.EVENT_TIME, resultMap.get(RawConstants.TIMESTAMP));
        }
        writeJson(resultMap);
        if (resultMap.containsKey(UPLOAD_SPEED)) {
            refreshConfig(resultMap, true);
            insertRawConn(resultMap);
        } else if (resultMap.containsKey("accX")) {
            insertRawTime(resultMap);
        }
    }

    /**
     * 测点统计包索引
     *
     * @param resultMap
     */
    void insertRawConn(Map<String, Object> resultMap) {
        String deviceCode = Convert.toStr(resultMap.get(RawConstants.CHIP_ID));
        List<String> pointList = checkBindPoint(deviceCode);
        if (pointList == null) return;
        Long ts = Convert.toLong(resultMap.get(RawConstants.TIMESTAMP));
        for (String point : pointList) {
            RawConn rawConn = new RawConn();
            rawConn.setTs(ts);
            rawConn.setEventTime(DateUtils.getTimestamp());
            rawConn.setDeviceCode(deviceCode);
            rawConn.setPointCode(point);
            rawService.insertRawConn(rawConn);
            log.info("RawConn storage completed  {}-{}-{}", point, deviceCode, ts);
        }
    }

    /**
     * 测点索引
     *
     * @param resultMap
     */
    void insertRawTime(Map<String, Object> resultMap) {
        String deviceCode = Convert.toStr(resultMap.get(RawConstants.CHIP_ID));
        List<String> pointList = checkBindPoint(deviceCode);
        //        String[] pointList = deviceCode.split(",");
        if (pointList == null) return;
        Long ts = Convert.toLong(resultMap.get(RawConstants.TIMESTAMP));
        for (String point : pointList) {
            RawTime rawTime = new RawTime();
            rawTime.setTs(ts);
            rawTime.setEventTime(DateUtils.getTimestamp());
            rawTime.setDeviceCode(deviceCode);
            rawTime.setPointCode(point);
            rawService.insertRawTime(rawTime);
            log.info("RawTime storage completed {}-{}-{}", point, deviceCode, ts);
            addRawGroup(rawTime);
        }
    }

    /**
     * 检测IoT绑定的测点列表
     *
     * @param deviceCode
     * @return
     */
    List<String> checkBindPoint(String deviceCode) {
        List<String> pointList = queryPointListByDevice(deviceCode);
        if (pointList == null) {
            log.warn("{} not bind point", deviceCode);
            return null;
        }
        return pointList;
    }

    /**
     * 刷新传感器配置
     *
     * @param resultMap
     * @param isGoodTs  时间戳是否正确
     */
    private void refreshConfig(Map<String, Object> resultMap, boolean isGoodTs) {
        RawConnDto rawConnDto = JSON.parseObject(JSON.toJSONString(resultMap), RawConnDto.class);
        // 2023-03-14 缓存传感器最新信息
        String infoKey = CacheConstants.RAW_SENSOR_INFO_KEY + rawConnDto.getChipID();
        Map<String, Object> mapInfo = redisService.getCacheMap(infoKey);
        if (mapInfo == null) {
            mapInfo = new HashMap<>(5);
        }
        mapInfo.put(RawConstants.EVENT_TIME, isGoodTs ? rawConnDto.getTimestamp() : DateUtils.getTimestamp());
        mapInfo.put(RawConstants.BATTERY_VOLTAGE, rawConnDto.getBatteryVoltage());
        mapInfo.put(RawConstants.WIFI_RSSI, rawConnDto.getWifiRssi());
        mapInfo.put(RawConstants.VERSION, rawConnDto.getVersion());
        if (rawConnDto.getConfigInfo() != null) {
            mapInfo.put(RawConstants.CONFIG_INFO, rawConnDto.getConfigInfo());
        }
        redisService.setCacheMap(infoKey, mapInfo);

        // 2023-03-24 检查传感器配置
        checkConfig(rawConnDto.getChipID(), rawConnDto.getVersion(), mapInfo.get(RawConstants.CONFIG_INFO));
    }

    /**
     * 传感器配置简写对照表
     */
    private static final HashMap<String, String> CFG_MAP = new HashMap<>() {{
        put("Syn", "SynTime");
        put("SPts", "SamplePoints");
        put("SItv", "SampleInterval");
        put("UCnt", "UploadFilesCount");
        put("FS", "VibFullScale");
        put("SUFg", "StartUseFlag");
        put("ReUF", "ReuploadFlag");
        put("Frq", "VibFrequency");
        put("RItv", "RmsRunItv");
        put("RTrd", "RmsThreshold");
        put("SPkg", "SubpackageSize");
        put("TMd", "TransferMode");
        put("SMd", "SampleMode");
    }};

    /**
     * 校验传感器配置是否最新，配置不全，自动下发MQTT
     *
     * @param chipId  传感器编码
     * @param version 版本号
     * @param config  配置信息
     */
    void checkConfig(String chipId, String version, Object config) {
        if (chipId == null || version == null) return;
        // 2023-01-04 判断版本是否需要升级
        String key = CacheConstants.IPHM_DEVICE_TASK + chipId;
        if (redisService.hasKey(key)) {
            Map<String, Object> map = redisService.getCacheObject(key);
            if (map.containsKey(RawConstants.VERSION) && map.get(RawConstants.VERSION).toString().equals(version)) {
                redisService.deleteObject(key);
                log.info("{} OTA upgrade successful，delete redis", chipId);
            } else {
                String json = JSON.toJSONString(map);
                sendSensorConfig(chipId, json);
                log.info("{} OTA upgrade failed，push OTA：{}", chipId, json);
            }
        }
        // 2023-03-24 版本2.2.1之后的配置变化后自动下发指令
        if (version.compareTo("2.2.1") > 0) {
            // 未有配置缓存，下发配置，重新上传
            if (config == null) {
                sendSensorConfig(chipId, null);
                log.info("{} sensor config cache does not exist，send config", chipId);
                return;
            }
            try {
                Map<String, Object> map = getSensorConfig(chipId);
                // 2023-05-30 传感器未绑定测点或配置信息为空，不下发配置
                if (map == null || map.isEmpty()) {
                    log.error("{} not found sensor config cache", chipId);
                    return;
                }
                Object obj;
                boolean isFloat, isOk;
                String[] configs = config.toString().split(",");
                for (String str : configs) {
                    // 获取上传的配置
                    String[] cfg = str.split(":");
                    // 2023-06-02 配置表中不存在的，不进行比较
                    if (!CFG_MAP.containsKey(cfg[0])) {
                        continue;
                    }
                    // rms阈值为浮点数
                    isFloat = "RTrd".equals(cfg[0]);
                    // 获取系统的配置
                    obj = map.get(CFG_MAP.get(cfg[0]));
                    if (isFloat) {
                        isOk = obj != null && Float.parseFloat(cfg[1]) == Float.parseFloat(obj.toString());
                    } else {
                        // 2023年12月06日 采样点数 取2的最大指数值
                        if ("SPts".equals(cfg[0])) {
                            isOk = obj != null && Integer.highestOneBit(Integer.parseInt(cfg[1])) == Integer.highestOneBit(Integer.parseInt(obj.toString()));
                        } else {
                            isOk = obj != null && Integer.parseInt(cfg[1]) == Integer.parseInt(obj.toString());
                        }
                    }
                    // 配置变更，下发新配置
                    if (!isOk) {
                        log.info("{} sensor config change [{}]，send config", chipId, cfg[0]);
                        sendSensorConfig(chipId, null);
                        return;
                    }
                }
            } catch (Exception ex) {
                log.error("{} sensor config check failed：", chipId, ex);
            }
        }
    }

    /**
     * 解析转速
     *
     * @param resultMap
     */
    private void unpackRPM(Map<String, Object> resultMap) {
        Long timestamp = validPackage(resultMap);
        if (timestamp == null) return;
        resultMap.replace(RawConstants.TIMESTAMP, timestamp);
        writeJson(resultMap);
    }

    /**
     * 将原始数据输出json文件
     *
     * @param resultMap 传感器编码
     */
    void writeJson(Map<String, Object> resultMap) {
        String dataType;
        if (resultMap.containsKey(UPLOAD_SPEED)) {
            dataType = "conn";
        } else if (resultMap.containsKey("Rpm")) {
            dataType = "rpm";
        } else if (resultMap.containsKey("velX")) {
            dataType = "vel";
        } else {
            dataType = "acc";
        }
        // 振动数据推送至第三方API
        if ("acc".equals(dataType) && transApi && StringUtils.isNotEmpty(transUrl)) {
            try {
                HttpUtils.sendPost(transUrl, JSON.toJSONString(resultMap));
                log.info("HTTP推送第三方API成功");
            } catch (Exception ex) {
                log.error("HTTP推送第三方API失败", ex);
            }
        }
        if (!write) return;
        Object chipId = resultMap.get(RawConstants.CHIP_ID);
        Long timestamp = Convert.toLong(resultMap.get(RawConstants.TIMESTAMP));
        String title = chipId + "-" + timestamp + " " + dataType;
        String folder = "raw/" + DateUtils.datePath(timestamp) + "/" + dataType + "/" + chipId;
        StopWatch watch = new StopWatch();
        watch.start();
        try {
            // 检查文件夹路径是否存在，不存在则创建
            Files.createDirectories(Paths.get(folder));
            // 构建目标文件路径
            String filePath = folder + "/" + timestamp + ".json";
            File file = new File(filePath);

            // 使用 try-with-resources 写入文件
            try (FileWriter writer = new FileWriter(file)) {
                writer.write(JSON.toJSONString(resultMap));
            }

            watch.stop();
            log.info("{} write json file successful. Elapsed time (ms) {}", title, watch.getTime());
        } catch (IOException e) {
            watch.stop();
            log.error("{} write json file failed. Elapsed time (ms): {}", title, watch.getTime(), e);
        }
    }

    /**
     * 按工作机组对测点攒包
     *
     * @param rawTime
     */
    synchronized void addRawGroup(RawTime rawTime) {
        // 推送测点数据至算法
        if (transMqtt) {
            mqttUtils.publishAsync(RawConstants.MQTT_DAQ, RawConstants.KAFKA_TRANS_RAW_TOPIC, JSON.toJSONString(rawTime));
            log.info("{}-{}-{} 测点数据推送算法", rawTime.getDeviceCode(), rawTime.getTs(), rawTime.getPointCode());
        }

        // 按工作机组进行攒包
        String pointGroup = rawTime.getPointCode().substring(0, 16);
        String key = CacheConstants.RAW_POINT_GROUP_KEY + pointGroup;
        String mapKey = rawTime.getTs().toString();
        List<RawTime> rawTimeList;
        Map<String, List<RawTime>> mapResult = redisService.getCacheMap(key);
        if (mapResult == null) {
            addHashMap(pointGroup, mapKey, rawTime, null);
            return;
        }
        // 加减60秒为同一组数据
        boolean isInclude = false;
        for (Map.Entry<String, List<RawTime>> entry : mapResult.entrySet()) {
            long timeDiff = Math.abs(Long.parseLong(entry.getKey()) - rawTime.getTs());
            if (timeDiff <= 60 * 1000 * 1000) {
                // 含有该分组
                isInclude = true;
                mapKey = entry.getKey();
                rawTimeList = entry.getValue();
                RawTime find = rawTimeList.stream().filter(l -> l.getPointCode().equals(rawTime.getPointCode())).findFirst().orElse(null);
                // 未有该测点，则加入排队中，且跳出比对
                if (find == null) {
                    addHashMap(pointGroup, mapKey, rawTime, rawTimeList);
                    break;
                }
            }
        }
        // 不包含，新创建一个key
        if (!isInclude) {
            addHashMap(pointGroup, mapKey, rawTime, null);
        }
        // 2023-04-13，整体性判断，前面数据包不够的进行等待
        checkRawGroup(pointGroup);
    }

    /**
     * HashMap 排队等待
     *
     * @param pointGroup  工作机组
     * @param mapKey      时间key
     * @param rawTime     测点数据
     * @param rawTimeList 时间key下测点数据列表
     */
    void addHashMap(String pointGroup, String mapKey, RawTime rawTime, List<RawTime> rawTimeList) {
        if (rawTimeList == null) {
            rawTimeList = new ArrayList<>();
            // 将key写入20分钟延迟过期
            redisService.setCacheObject(CacheConstants.RAW_POINT_GROUP_EXPIRE_KEY + pointGroup + "-" + mapKey, mapKey, waitTimeout, TimeUnit.MINUTES);
        }
        // 入队等待
        rawTimeList.add(rawTime);
        redisService.setCacheMapValue(CacheConstants.RAW_POINT_GROUP_KEY + pointGroup, mapKey, rawTimeList);
        // 设置过期时间
        redisService.expire(CacheConstants.RAW_POINT_GROUP_KEY + pointGroup, 1L, TimeUnit.DAYS);
        log.info("工作机组 " + pointGroup + " -- " + rawTime.getPointCode() + "-" + rawTime.getTs() + " 入队 " + mapKey + " 长度 " + rawTimeList.size());
    }

    /**
     * 对工作机组整体进行排序转发
     *
     * @param pointGroup
     */
    public synchronized void checkRawGroup(String pointGroup) {
        String key = CacheConstants.RAW_POINT_GROUP_KEY + pointGroup;
        log.info("工作组机 {} 检测攒包", key);
        // 取出某工作机组下的攒包列表
        Map<String, List<RawTime>> result = redisService.getCacheMap(key);
        // 取出工作机组下测点列表
        List<String> pointList = queryPointListByGroup(pointGroup);
        int size = pointList != null ? pointList.size() : 0;
        // 将 HashMap 转换为 TreeMap 自然顺序
        Long ts = DateUtils.getTimestamp();
        TreeMap<String, List<RawTime>> treeMap = new TreeMap<>(result);
        // 2023-04-13 整体性判断，排序后循环往下执行，直到待未过期的
        for (Map.Entry<String, List<RawTime>> entry : treeMap.entrySet()) {
            String expireKey = CacheConstants.RAW_POINT_GROUP_EXPIRE_KEY + pointGroup + "-" + entry.getKey();
            // 2024年01月19日 key失效且超时，直接转发，防止丢失超时key，导致无法转发机组索引
            if (!redisService.hasKey(expireKey) && DateUtils.getHoursBetween(Convert.toLong(entry.getKey()), ts) >= waitTimeout) {
                log.info("工作机组等待超时，开始转发 {}", pointGroup + "-" + entry.getKey());
            } else {
                // 第一包数据长度不够，跳出，不进行转发
                if (entry.getValue().size() != size) return;
                log.info("工作机组攒包成功，开始转发 {}", pointGroup + "-" + entry.getKey());
            }
            // 删除攒包缓存
            redisService.deleteCacheMapValue(key, entry.getKey());
            // 转发数据
            pushRawGroup(entry.getValue());
        }
    }

    /**
     * 工作机组数据推送算法并入库
     *
     * @param rawTimeList
     */
    private void pushRawGroup(List<RawTime> rawTimeList) {
        RawTime first = rawTimeList.get(0);
        String equipmentCode = getEquipmentCodeByPoint(first.getPointCode());
        if (equipmentCode == null) {
            log.error("{} 测点所在工作机组查询失败", first.getPointCode());
            return;
        }
        RawGroup rawGroup = new RawGroup();
        rawGroup.setTimestamp(first.getTs());
        rawGroup.setGroupCode(equipmentCode);
        rawGroup.setPointList(JSON.toJSONString(rawTimeList));
        rawService.insertRawGroup(rawGroup);
        log.info("RawGroup storage completed {}-{}", equipmentCode, first.getTs());

        Map<String, Object> map = new HashMap<>(3);
        map.put("ts", first.getTs());
        map.put("equipmentCode", equipmentCode);
        map.put("pointList", rawTimeList);
        if (transMqtt) {
            mqttUtils.publishAsync(RawConstants.MQTT_DAQ, RawConstants.KAFKA_TRANS_RAW_GROUP_TOPIC, JSON.toJSONString(map));
            log.info("{}-{} 长度 {} 工作机组攒包推送算法", equipmentCode, first.getTs(), rawTimeList.size());
        }
    }

    /**
     * 获取传感器配置
     *
     * @param chipId 传感器编码
     * @return
     */
    Map<String, Object> getSensorConfig(String chipId) {
        List<Device> deviceList = redisService.getCacheList(CacheConstants.EDGE_SENSOR_LIST);
        if (StringUtils.isEmpty(deviceList)) {
            log.error("Not found device list cache");
            return null;
        }
        Device find = deviceList.stream().filter(l -> l.getDeviceCode().equals(chipId)).findFirst().orElse(null);
        return find == null ? null : DataFormatter.formatMap(JSON.parseObject(find.getConfig()));
    }

    /**
     * 下发传感器配置
     *
     * @param chipId 传感器编码
     * @param config 配置信息
     */
    void sendSensorConfig(String chipId, String config) {
        if (config == null) {
            Map<String, Object> map = getSensorConfig(chipId);
            if (map == null) return;
            config = JSON.toJSONString(map);
        }
        if (StringUtils.isNotEmpty(config)) {
            mqttUtils.publishAsync(RawConstants.MQTT_DAQ, "cfg/" + chipId, config);
        }
    }

    /**
     * 按测点查询出所在工作机组第一个设备
     *
     * @param pointCode
     * @return
     */
    String getEquipmentCodeByPoint(String pointCode) {
        List<Equipment> equipmentList = redisService.getCacheList(CacheConstants.EDGE_EQUIP_LIST);
        if (StringUtils.isEmpty(equipmentList)) {
            log.error("Not found equip list cache");
            return null;
        }
        try {
            Equipment find = equipmentList.stream().filter(l -> l.getLevel() > 10 && l.getEquipmentCode().substring(0, 16).equals(pointCode.substring(0, 16))).findFirst().orElse(null);
            return find == null ? null : find.getEquipmentCode();
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 取出工作机组下测点列表
     *
     * @param pointGroup 工作机组编码（测点前16位编码）
     * @return
     */
    List<String> queryPointListByGroup(String pointGroup) {
        List<Point> pointList = getPointList();
        return pointList == null ? null : pointList.stream().map(Point::getPointCode).filter(pointCode -> pointCode.contains(pointGroup)).collect(Collectors.toList());
    }

    /**
     * 按IoT编码查询测点列表
     *
     * @param deviceCode
     * @return 无数据会返回 一个空的列表 []
     */
    List<String> queryPointListByDevice(String deviceCode) {
        List<Point> pointList = getPointList();
        return pointList == null ? null : pointList.stream().filter(l -> l.getDeviceCode() != null && l.getDeviceCode().equals(deviceCode)).map(Point::getPointCode).collect(Collectors.toList());
    }

    /**
     * 查询测点列表，过滤掉没有绑定传感器的
     *
     * @return
     */
    List<Point> getPointList() {
        List<Point> pointList = redisService.getCacheList(CacheConstants.EDGE_POINT_LIST);
        if (StringUtils.isEmpty(pointList)) {
            log.error("Not found point list cache");
            return null;
        }
        return pointList.stream().filter(l -> StringUtils.isNotEmpty(l.getDeviceCode())).toList();
    }
}